package com.rwind.function;

import com.rwind.model.Event;
import com.rwind.source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Timestamp;

/**
 * 窗口处理函数
 * <pre>
 * 基于窗口的ProcessWindowFunction和ProcessAllWindowFunction
 * </pre>
 * <pre>
 * 侧输出流
 * 这里output()方法需要传入两个参数，第一个是一个“输出标签”OutputTag，用来标识侧输出流，一般会在外部统一声明；第二个就是要输出的数据。
 * OutputTag<String> outputTag = new OutputTag<String>("side-output")
 * 如果想要获取这个侧输出流，可以基于处理之后的DataStream直接调用.getSideOutput()方法，传入对应的OutputTag，这个方式与窗口API中获取侧输出流是完全一样的。
 * DataStream<String> stringStream = longStream.getSideOutput(outputTag);
 * </pre>
 */
public class MyKeyedProcessFunction extends KeyedProcessFunction<Boolean, Event, String> {

    OutputTag<String> outputTag = new OutputTag<String>("side-output") {
    };

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 处理时间语义，不需要分配时间戳和 watermark
        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource());
        stream.assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event element, long recordTimestamp) {
                        return element.timestamp;
                    }
                }));
        //keyBy--分区键类型
        //是将所有数据的key 都指定为了true，其实就是所有数据拥有相同的key，会分配到同一个分区。
        stream.keyBy(data -> true).process(new MyKeyedProcessFunction()).print();
        //stream.print();
        env.execute();
    }

    @Override
    /**
     * Boolean --> 分区键类型
     * Event--->事件数据类型
     * String--->产生新数据类型
     * 每来一个数据都会调用一次，主要是定义了一个 10 秒之后的定时器；而.onTimer()方法则会在定时器触发时调用。
     */
    public void processElement(Event value, KeyedProcessFunction<Boolean, Event, String>.Context ctx, Collector<String> out) throws Exception {
        //获取当前数据的key
        Boolean currentKey = ctx.getCurrentKey();

        // 定时器注册
        TimerService timerService = ctx.timerService();

        // 1、事件时间的案例
        // 数据中提取出来的事件时间
        Long currentEventTime = ctx.timestamp();
        timerService.registerEventTimeTimer(5000L);
        System.out.println("当前key=" + currentKey + ",当前时间=" + currentEventTime + ",注册了一个5s的定时器");

        // 获取当前时间的watermark
        long currentWatermark = timerService.currentWatermark();
        System.out.println("当前数据=" + value + ",当前watermark=" + currentWatermark);


        long currTs = ctx.timerService().currentProcessingTime();
        out.collect("数据到达，到达时间：" + new Timestamp(currTs));
        out.collect(" 数 据 到 达 ， 水 位 线 为： " + ctx.timerService().currentWatermark() + " -------分割线 \n");
        // 注册一个 10秒后的定时器
        ctx.timerService().registerProcessingTimeTimer(currTs + 10 * 1000L);

        // 转换成String，输出到侧输出流中
        ctx.output(outputTag, "side-output: " + String.valueOf(value));

        // 注册一个10秒后的定时器
       // ctx.timerService().registerEventTimeTimer(ctx.timestamp()+10*1000L);

        //注册定时器
        //ctx.timerService().registerProcessingTimeTimer(ts);
        //取消定时器注册
        //ctx.timerService().deleteProcessingTimeTimer(timerTS);

        /**
         * // 删除触发时间为time的事件时间定时器
         * registerEventTimeTimer();
         *
         * // 删除触发时间为time的事件时间定时器
         * void deleteEventTimeTimer(long time);
         * 最多只有一个定时器，如果注册了多次，onTimer()方法也将只被调用一次。
         */

    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<Boolean, Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
        out.collect("定时器触发，触发时间：" + new Timestamp(timestamp));

    }
}
